package com.iussoft.receiveData.send.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.tools.DateUtils;
import com.apache.tools.StrUtil;
import com.iussoft.model.DataParams;
import com.iussoft.receiveData.redis.EsManager;
import com.iussoft.tools.AesUtil;
import com.iussoft.tools.HttpService;
import com.iussoft.tools.LoggerUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * @description: 存证阶段
 * @author: Hou Dayu
 * @date: Created in 2020/4/1
 */
@Service("depositCertificateManager")
public class DepositCertificateManager extends SuperSearchManager {

    @Autowired
    @Qualifier("esManager")
    protected EsManager esManager;

    /**
     * 同步数据获取方式实现方法
     */
    public void sendDataMsg(DataParams params) {
        LoggerUtil.info("数据存证阶段开始...");
        String rowkey = rollbackStr(params.getTraceId());
        String sourceMsg = params.getParamsToStr("sourceMsg");
        String interfaceId = "";
        String infterfacePwd = "";
        String interfaceCode = "";
        if(StrUtil.isNotNull(params.getParamsToStr("interfaceInfo"))) {
            JSONObject obj = JSON.parseObject(params.getParamsToStr("interfaceInfo"));
            interfaceId = StrUtil.doNull(obj.getString("infoId"), "");
            infterfacePwd = StrUtil.doNull(obj.getString("infterfacePwd"), "");
            interfaceCode = StrUtil.doNull(obj.getString("interfaceCode"), "");
        }
        Map<String,Object> esinfo = new HashMap();
        esinfo.put("infoId", interfaceId);
        esinfo.put("traceId", params.getParamsToStr("traceId"));
        esinfo.put("rowkey", rowkey);
        esinfo.put("sourceCode", StrUtil.doNull(params.getParamsToStr("requestCode"), ""));
        esinfo.put("interfaceCode", interfaceCode);
        esinfo.put("infterfacePwd", infterfacePwd);
        esinfo.put("flag", "send");
        esinfo.put("createdTime", DateUtils.strToLong(DateUtils.Now.fmt_yyyyMMdd_HHmmss(), DateUtils.FmtStr.yyyyMMdd_HHmmss));
        boolean ifKafka = kafkaSimpleSender.getKafkaEnabled();
        String index_name = analyzeIndexName + "-" + DateUtils.Now.fmt_yyyy();
        if(ifKafka) {
            String reqHbaseUrl = esManager.getRequestUrl("search-server");
            esinfo.put("tableName",getHbaseTableName(1));
            String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/type=string";
            String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(sourceMsg));
            HttpService.sendHttp(url, "post", null, msg, "json");
        }else {
            esinfo.put("message",sourceMsg);
            esinfo.put("tableName", index_name);
        }
        boolean mark = esManager.saveInfoForEs(indexName+"-"+DateUtils.Now.fmt_yyyy(), indexName, rowkey, esinfo);
        LoggerUtil.warnSource("traceId = {}; exeSys = send; infoType = info; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}",params.getParamsToStr("traceId"), "deposit_certificate",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
    }

    /**
     * 同步通道方式实现方法
     */
    public void syncSendData(DataParams params) {
        LoggerUtil.info("数据存证阶段开始...");
        String rowkey = rollbackStr(params.getTraceId());
        String sourceMsg = params.getParamsToStr("sourceMsg");
        String interfaceId = "";
        String infterfacePwd = "";
        String interfaceCode = "";
        if(StrUtil.isNotNull(params.getParamsToStr("interfaceInfo"))) {
            JSONObject obj = JSON.parseObject(params.getParamsToStr("interfaceInfo"));
            interfaceId = StrUtil.doNull(obj.getString("infoId"), "");
            infterfacePwd = StrUtil.doNull(obj.getString("infterfacePwd"), "");
            interfaceCode = StrUtil.doNull(obj.getString("interfaceCode"), "");
        }
        Map<String,Object> esinfo = new HashMap();
        esinfo.put("infoId", interfaceId);
        esinfo.put("traceId", params.getParamsToStr("traceId"));
        esinfo.put("rowkey", rowkey);
        esinfo.put("sourceCode", StrUtil.doNull(params.getParamsToStr("requestCode"), ""));
        esinfo.put("interfaceCode", interfaceCode);
        esinfo.put("infterfacePwd", infterfacePwd);
        esinfo.put("flag", "sync");
        esinfo.put("createdTime", DateUtils.strToLong(DateUtils.Now.fmt_yyyyMMdd_HHmmss(), DateUtils.FmtStr.yyyyMMdd_HHmmss));
        boolean ifKafka = kafkaSimpleSender.getKafkaEnabled();
        String index_name = analyzeIndexName + "-" + DateUtils.Now.fmt_yyyy();
        if(ifKafka) {
            String reqHbaseUrl = esManager.getRequestUrl("search-server");
            esinfo.put("tableName",getHbaseTableName(1));
            String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/type=string";
            String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(sourceMsg));
            HttpService.sendHttp(url, "post", null, msg, "json");
        }else {
            esinfo.put("message",sourceMsg);
            esinfo.put("tableName", index_name);
        }
        boolean mark = esManager.saveInfoForEs(indexName+"-"+DateUtils.Now.fmt_yyyy(), indexName, rowkey, esinfo);
        LoggerUtil.warnSource("traceId = {}; exeSys = send; infoType = info; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}",params.getParamsToStr("traceId"), "deposit_certificate",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
    }

    /**
     * 数据接入方式实现方法
     */
    public void receiveDataMsg(DataParams params) {
        LoggerUtil.info("数据存证阶段开始...");
        String rowkey = rollbackStr(params.getTraceId());
        String interfaceCode = params.getParamsToStr("interfaceCode");
        String sourceMsg = params.getParamsToStr("sourceMsg");
        String interfaceId = StrUtil.doNull(params.getParamsToStr("infterfaceId"), "");
        String infterfacePwd = StrUtil.doNull(params.getParamsToStr("infterfacePwd"), "");
        Map<String,Object> esinfo = new HashMap();
        esinfo.put("infoId", interfaceId);
        esinfo.put("traceId", params.getParamsToStr("traceId"));
        esinfo.put("rowkey", rowkey);
        esinfo.put("sourceCode", StrUtil.doNull(params.getParamsToStr("requestCode"), ""));
        esinfo.put("interfaceCode", interfaceCode);
        esinfo.put("infterfacePwd", infterfacePwd);
        esinfo.put("flag", "receive");
        esinfo.put("createdTime", DateUtils.strToLong(DateUtils.Now.fmt_yyyyMMdd_HHmmss(), DateUtils.FmtStr.yyyyMMdd_HHmmss));
        boolean ifKafka = kafkaSimpleSender.getKafkaEnabled();
        String index_name = analyzeIndexName + "-" + DateUtils.Now.fmt_yyyy();
        if(ifKafka) {
            String reqHbaseUrl = esManager.getRequestUrl("search-server");
            esinfo.put("tableName",getHbaseTableName(1));
            String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/type=string";
            String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(sourceMsg));
            HttpService.sendHttp(url, "post", null, msg, "json");
        }else {
            esinfo.put("message",sourceMsg);
            esinfo.put("tableName", index_name);
        }
        boolean mark = esManager.saveInfoForEs(indexName+"-"+DateUtils.Now.fmt_yyyy(), indexName, rowkey, esinfo);
        LoggerUtil.warnSource("traceId = {}; exeSys = send; infoType = info; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}",params.getParamsToStr("traceId"), "deposit_certificate",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
    }

    /**
     * 异步数据获取方式实现方法
     */
    public void asyncSendData(DataParams params) {
        LoggerUtil.info("数据存证阶段开始...");
        String rowkey = rollbackStr(params.getTraceId());
        String interfaceCode = params.getParamsToStr("interfaceCode");
        String sourceMsg = params.getParamsToStr("sourceMsg");
        String interfaceId = StrUtil.doNull(params.getParamsToStr("infterfaceId"), "");
        String infterfacePwd = StrUtil.doNull(params.getParamsToStr("infterfacePwd"), "");
        Map<String,Object> esinfo = new HashMap();
        esinfo.put("infoId", interfaceId);
        esinfo.put("traceId", params.getParamsToStr("traceId"));
        esinfo.put("rowkey", rowkey);
        esinfo.put("sourceCode", StrUtil.doNull(params.getParamsToStr("requestCode"), ""));
        esinfo.put("interfaceCode", interfaceCode);
        esinfo.put("infterfacePwd", infterfacePwd);
        esinfo.put("flag", "asyncSend");
        esinfo.put("createdTime", DateUtils.strToLong(DateUtils.Now.fmt_yyyyMMdd_HHmmss(), DateUtils.FmtStr.yyyyMMdd_HHmmss));
        boolean ifKafka = kafkaSimpleSender.getKafkaEnabled();
        String index_name = analyzeIndexName + "-" + DateUtils.Now.fmt_yyyy();
        if(ifKafka) {
            String reqHbaseUrl = esManager.getRequestUrl("search-server");
            esinfo.put("tableName",getHbaseTableName(1));
            String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/type=string";
            String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(sourceMsg));
            HttpService.sendHttp(url, "post", null, msg, "json");
        }else {
            esinfo.put("message",sourceMsg);
            esinfo.put("tableName", index_name);
        }
        boolean mark = esManager.saveInfoForEs(indexName+"-"+DateUtils.Now.fmt_yyyy(), indexName, rowkey, esinfo);
        LoggerUtil.warnSource("traceId = {}; exeSys = send; infoType = info; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}",params.getParamsToStr("traceId"), "deposit_certificate",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
    }
}
